package com.zhanghe.study.study_scala.akka.spark.master

import java.util.concurrent.TimeUnit

import akka.actor.{Actor, ActorSystem, Props}
import com.typesafe.config.ConfigFactory
import com.zhanghe.study.study_scala.akka.spark.common.{HeartBeat, RegisterWorkerInfo, RegisteredWorkerInfo, RemoveTimeOutWorker, SendHeartBeat, StartTimeOutWorker, WorkerInfo}

import scala.concurrent.duration.FiniteDuration

/**
 * @author zh
 * @date 2021/5/10 15:51
 */
object SparkMaster {
  def main(args: Array[String]): Unit = {
    // 创建ActorSystem  remote配置
    // 注意：想要发送对象的话需要进行序列化配置 akka.actor.allow-java-serialization=on
    val config = ConfigFactory.parseString(
      s"""
         |akka.actor.provider="akka.remote.RemoteActorRefProvider"
         |akka.remote.artery.canonical.hostname=127.0.0.1
         |akka.remote.artery.canonical.port=10005
         |akka.actor.allow-java-serialization=on
            """.stripMargin)
    val sparkMasterSystem = ActorSystem("SparkMaster", config)
    // 创建sparkMaster
    val sparkMasterRef = sparkMasterSystem.actorOf(Props[SparkMaster],"SparkMaster")

    // 启动
    sparkMasterRef.!("start")
  }

}

class SparkMaster extends Actor{

  import scala.collection.mutable
  // 定义hashmap  管理worker信息
  val workers = mutable.Map[String,WorkerInfo]()

  override def receive: Receive = {
    case "start" => {
      println("sparkMaster启动成功了")
      self.!(StartTimeOutWorker)
    }
    case RegisterWorkerInfo(id,cpu,ram) =>{
      if(!workers.contains(id)){
        // 组织workerInfo信息
        val workerInfo = new WorkerInfo(id,cpu,ram)
        workers.put(id,workerInfo)
      }
      println(s"worker-$id 注册成功")
      sender().!(RegisteredWorkerInfo)

    }

    case HeartBeat(id) =>{
      // 更新心跳时间
      if(workers.contains(id)){
        val workerInfo = workers(id)
        workerInfo.lastHeartBeat = System.currentTimeMillis()
        println(s"master更新 $id 心跳时间")
      }
    }

    case StartTimeOutWorker => {
      println("开始定时检测worker心跳任务")
      // 注册成功之后，启动定时器，每隔一段时间检测超时的worker
      import context.dispatcher
      context.system.scheduler.scheduleAtFixedRate(FiniteDuration(0,TimeUnit.MILLISECONDS),FiniteDuration(9000,TimeUnit.MILLISECONDS),self,RemoveTimeOutWorker)
    }

    case RemoveTimeOutWorker => {
      // 如果心跳超时，从map中删除
      val nowTime = System.currentTimeMillis()
      val workerInfos = workers.values
      workerInfos.filter(workerInfo => (nowTime - workerInfo.lastHeartBeat)>6000)
        .foreach(workerInfo => workers.remove(workerInfo.id))
      println(s"当前还有 ${workers.size} 个worker存活")
    }
  }
}
